from kafka import KafkaConsumer
from multiprocessing import Process, current_process
import time
import json
import requests
from datetime import datetime
import pymysql

#数据清洗和提取，网站get请求，数据库写入

conn = pymysql.connect(
    host="192.168.138.99",
    user="yang",
    passwd="123456",
    port=3306,
    db="log_consumer"
)
cur = conn.cursor()

def consume_kafka_partition(topic, group_id, partition):
    """
    进程执行的函数，用于消费指定分区的Kafka消息
    """
    consumer = KafkaConsumer(
        group_id=group_id,
        bootstrap_servers=['192.168.138.88:9092','192.168.138.77:9092','192.168.138.99:9092',],
        auto_offset_reset='earliest',  # 可按需调整偏移量重置策略
        enable_auto_commit=True,       # 自动提交偏移量
        auto_commit_interval_ms=5000,  # 每5秒提交一次偏移量
        value_deserializer=lambda x: x.decode('utf-8')  # 将消息值解码为UTF-8字符串
    )
    # 为消费者订阅的主题
    consumer.subscribe(topic)

    for message in consumer:
        # print(f"message:{message}")
        # print(f"进程 {current_process().name} 消费到来自分区 {partition} 的消息: {message.value}")
        message_str = json.loads(message.value)
        message_log = message_str['message']
        message_log_lst = message_log.split(' ')
        # print(message_log_lst[0], message_log_lst[3], message_log_lst[9])

        # url = "https://whois.pconline.com.cn/ipJson.jsp?ip=114.114.114.114&json=true"
        # response = requests.get(url)
        # response.raise_for_status()  # 检查请求是否成功
        # data = response.json()  # 解析JSON响应
        # print(data)
        data_dic =  {'ip': '114.114.114.114', 'pro': '江苏省', 'proCode': '320000', 'city': '南京市', 'cityCode': '320100',
          'region': '', 'regionCode': '0', 'addr': '江苏省南京市 电信', 'regionNames': '', 'err': ''}

        log_ip = message_log_lst[0]
        log_city = data_dic['addr'].split(' ')[0]
        log_operator = data_dic['addr'].split(' ')[1]
        log_flow = message_log_lst[9]
        log_date_str = message_log_lst[3]
        date_str = log_date_str[1:12]
        #datetime.strptime() 是用来解析日期字符串并将其转化为 datetime 对象的方法。
        date_obj = datetime.strptime(date_str, "%d/%b/%Y")
        #"%d/%b/%Y" 是日期格式的指示符，用来告诉 Python 如何解析 date_str 字符串：%b 表示月份的英文简写
        formatted_date = date_obj.strftime("%Y%m%d")
        result = []
        result.append(log_ip)
        result.append(log_city)
        result.append(log_operator)
        result.append(formatted_date)
        result.append(log_flow)
        print(formatted_date)
        sql = "INSERT INTO log values(%s,%s,%s,%s,%s)"
        cur.execute(sql, result)
        conn.commit()


if __name__ == "__main__":

    topic = "nginxlog"  # 根据实际情况修改要消费的主题名称
    group_id = "message_group3"  # 根据实际情况修改消费组名称
    partitions = [0, 1, 2]  # 定义要消费的分区列表，可按需调整

    # 设置进程名称（可选）
    def start_process(target, args):
        p = Process(target=target, args=args)
        p.name = f"Consumer-Partition-{args[2]}"  # 设置进程名称
        p.start()
        return p

    processes = []
    for partition in partitions:
        p = start_process(consume_kafka_partition, (topic, group_id, partition))
        processes.append(p)

    for p in processes:
        p.join()

    print("所有进程已结束，指定分区消费完成")